package com.example.artemis.abstracts;

import com.example.artemis.config.AbstractProducerConfig;
import com.example.artemis.events.EventBus;
import com.example.artemis.events.MessageSentEvent;
import com.example.artemis.interfaces.MessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.io.Serializable;

/**
 * 基础消息生产者抽象类
 * 提供线程安全的消息发送功能
 * 修复了并发Session使用问题
 */
public abstract class AbstractMessageProducer implements MessageProducer {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    protected final AbstractProducerConfig config;
    protected final EventBus eventBus;

    // 使用ThreadLocal确保每个线程有独立的JMS资源
    private final ThreadLocal<JMSResources> threadLocalResources = new ThreadLocal<>();

    // 共享的Connection（线程安全）
    protected Connection connection;

    public AbstractMessageProducer(AbstractProducerConfig config) {
        this.config = config;
        this.eventBus = EventBus.getInstance();
        initializeSharedResources();
    }

    /**
     * 初始化共享的JMS连接资源
     */
    private void initializeSharedResources() {
        try {
            ConnectionFactory connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(config.getConnectorUrl());
            connection = connectionFactory.createConnection();
            connection.start();

            logger.info("共享JMS连接初始化成功");
        } catch (JMSException e) {
            logger.error("初始化JMS连接失败", e);
            throw new RuntimeException("初始化JMS连接失败", e);
        }
    }

    /**
     * JMS资源封装类，包含Session和Producer
     */
    private static class JMSResources {
        final Session session;
        final javax.jms.MessageProducer producer;

        JMSResources(Connection connection, AbstractProducerConfig config) throws JMSException {
            this.session = connection.createSession(config.isTransacted(), config.getSessionMode());
            this.producer = session.createProducer(null);
            this.producer.setDeliveryMode(config.isPersistent() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
            this.producer.setPriority(config.getMessagePriority());
            this.producer.setTimeToLive(config.getMessageTimeToLive());
        }

        void close() {
            try {
                if (producer != null) {
                    producer.close();
                }
                if (session != null) {
                    session.close();
                }
            } catch (JMSException e) {
                // 静默关闭错误
            }
        }
    }

    /**
     * 获取当前线程的JMS资源，如果不存在则创建
     */
    private JMSResources getCurrentThreadResources() throws JMSException {
        JMSResources resources = threadLocalResources.get();
        if (resources == null) {
            resources = new JMSResources(connection, config);
            threadLocalResources.set(resources);
            logger.debug("为线程 {} 创建新的JMS资源", Thread.currentThread().getName());
        }
        return resources;
    }

    @Override
    public void sendMessage(String queueName, Serializable message) throws Exception {
        long startTime = System.currentTimeMillis();
        String messageId = "msg-" + System.currentTimeMillis() + "-" + message.hashCode();

        JMSResources resources = null;
        try {
            // 获取当前线程的JMS资源
            resources = getCurrentThreadResources();

            Queue queue = resources.session.createQueue(queueName);

            ObjectMessage jmsMessage = resources.session.createObjectMessage();
            jmsMessage.setObject(message);

            // 设置消息属性
            jmsMessage.setIntProperty("retryCount", 0);
            jmsMessage.setStringProperty("queueName", queueName);
            jmsMessage.setLongProperty("messageCreatedTime", System.currentTimeMillis());

            resources.producer.send(queue, jmsMessage);

            // 发布消息发送成功事件
            long productionTime = System.currentTimeMillis() - startTime;
            int messageSize = calculateMessageSize(message);
            String producerId = getProducerId();

            publishMessageSentEvent(messageId, queueName, producerId, productionTime, messageSize);

            logger.debug("消息发送成功 - 队列: {}, 消息ID: {}", queueName, jmsMessage.getJMSMessageID());
        } catch (JMSException e) {
            // 发布消息发送失败事件
            long productionTime = System.currentTimeMillis() - startTime;
            String producerId = getProducerId();

            logger.error("消息发送失败 - 队列: {}, 错误: {}, 耗时: {}ms", queueName, e.getMessage(), productionTime);

            handleSendException(e, message, queueName);
            throw e;
        }
    }

    /**
     * 处理发送异常
     */
    protected void handleSendException(JMSException e, Serializable message, String queueName) {
        logger.error("消息发送失败 - 队列: {}, 错误: {}", queueName, e.getMessage(), e);
    }

    /**
     * 发布消息发送成功事件
     */
    protected void publishMessageSentEvent(String messageId, String queueName, String producerId,
                                         long productionTime, int messageSize) {
        try {
            MessageSentEvent event = new MessageSentEvent(messageId, queueName, producerId, productionTime, messageSize);
            eventBus.publishAsync(event);
            logger.debug("发布消息发送成功事件: {}", messageId);
        } catch (Exception e) {
            logger.error("发布消息发送成功事件时发生错误", e);
        }
    }

    /**
     * 获取生产者ID
     */
    protected String getProducerId() {
        return "producer-" + this.getClass().getSimpleName();
    }

    /**
     * 计算消息大小（简单估算）
     */
    protected int calculateMessageSize(Serializable message) {
        if (message == null) {
            return 0;
        }
        try {
            // 简单的消息大小估算，实际项目中可能需要更精确的计算
            return message.toString().getBytes().length;
        } catch (Exception e) {
            logger.debug("计算消息大小时发生错误", e);
            return 100; // 默认大小
        }
    }

    /**
     * 关闭资源
     */
    public void close() {
        // 关闭当前线程的JMS资源
        closeCurrentThreadResources();

        // 注意：ThreadLocal 没有简单的方法来访问所有线程的资源
        // 线程安全的资源清理依赖于各个线程在结束时调用 closeCurrentThreadResources()

        // 关闭共享连接
        try {
            if (connection != null) {
                connection.close();
            }
            logger.info("消息生产者资源已关闭");
        } catch (JMSException e) {
            logger.error("关闭JMS连接时发生错误", e);
        }
    }

    /**
     * 关闭当前线程的JMS资源
     * 在线程结束时调用以避免资源泄漏
     */
    public void closeCurrentThreadResources() {
        JMSResources resources = threadLocalResources.get();
        if (resources != null) {
            resources.close();
            threadLocalResources.remove();
            logger.debug("当前线程的JMS资源已关闭: {}", Thread.currentThread().getName());
        }
    }

    /**
     * 根据消息计算队列名称（可选方法）
     */
    protected String calculateQueueName(Serializable message) {
        int queueIndex = Math.abs(message.hashCode() % config.getQueueCount());
        return config.getQueueNamePrefix() + "." + queueIndex;
    }
}